package net.oschina.zbus;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.httpclient.methods.InputStreamRequestEntity;
import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zbus.client.Broker;
import org.zbus.client.broker.SingleBroker;
import org.zbus.client.broker.SingleBrokerConfig;
import org.zbus.client.service.ServiceConfig;
import org.zbus.client.service.ServiceHandler;
import org.zbus.client.service.ServiceLoader;
import org.zbus.client.service.ServiceProvider;
import org.zbus.remoting.Message;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
//本地启动zbus, 在test目录中ZbusStarter可内嵌式启动zbus服务器
/**
 * 
 * 发行时，可选择编译成jar放在zbus启动参数-serviceBase指定的目录下任意位置, zbus自动加载
 * zbus加载内嵌服务要求：
 * 1） 实现了ServiceProvider接口
 * 2） 提供无参数构造函数
 * 
 * @author 洪磊明(rushmore)
 *
 */
public class HttpCallbackService implements ServiceProvider, ServiceHandler {
	private static final Logger log = LoggerFactory.getLogger(HttpCallbackService.class);
	
	private ExecutorService executor = new ThreadPoolExecutor(4, 16, 120, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
	private final Properties props = new Properties();
	private final static String CONFIG_FILE = "HttpCallback.properties";
	 
	public HttpCallbackService(){
		//从配置文件中读取配置信息
		InputStream stream = getClass().getClassLoader().getResourceAsStream(CONFIG_FILE);
		try {
			if(stream != null){
				props.load(stream);
			} else {
				log.warn("missing properties: "+ CONFIG_FILE);
			}
		} catch (IOException e) {
			log.error(e.getMessage(), e);
		}
	}
	
	@Override
	public ServiceConfig getConfig() { 
		ServiceConfig config = new ServiceConfig();
		String mq = props.getProperty("mqName", "HttpCallback").trim();
		String registerToken = props.getProperty("registerToken", "").trim();
		String accessToken = props.getProperty("accessToken", "").trim();
		String consumerCountStr = props.getProperty("consumerCount", "1").trim();
		int consumerCount = 1;
		try{
			consumerCount = Integer.valueOf(consumerCountStr); 
		} catch (Exception e){
			log.error(e.getMessage(), e); 
		}
		
		config.setMq(mq); 
		config.setRegisterToken(registerToken);
		config.setAccessToken(accessToken);
		config.setThreadCount(Integer.valueOf(consumerCount));
		config.setServiceHandler(this);
		return config;
	}
	
	
	public void handleCallbackRequest(Request req){
		log.info(req.toString());
		//填写业务逻辑
		String dest_url = req.getDest_url();
		String dest_body = req.getDest_body();
		String token_url = req.getToken_url();
		int retry_count = req.getRetry_count();
		InputStream in = null;
		Response response = null;
		try{
			for(int i=0; i<retry_count; i++){
				//推送目标信息到目标地址
				in = IOUtils.toInputStream(dest_body, "UTF-8");
				InputStreamRequestEntity entity = new InputStreamRequestEntity(in, in.available(), "application/json");
				response = HttpKit.post(dest_url, entity);
				if(response != null){
					//推送返回信息到回掉地址
					String responseJSON = JSONObject.toJSONString(response);
					in = IOUtils.toInputStream(responseJSON);
					entity = new InputStreamRequestEntity(in, in.available(), "application/json");
					HttpKit.post(token_url, entity);
					break;
				}
			}
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	
	
	@Override
	public Message handleRequest(Message msg) { 
		try{
			final Request req = JSON.parseObject(msg.getBodyString(), Request.class);
			executor.submit(new Runnable() {
				@Override
				public void run() { 
					handleCallbackRequest(req);
				}
			});
			
		} catch (Throwable e){
			log.error(e.getMessage(), e);
		}
		return null;
	}
	
	
	/**
	 * 单独测试注册到指定的zbus，插件模式下只需要把本class所在的jar放到zbus插件可扫描的任何目录下即可
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		SingleBrokerConfig config = new SingleBrokerConfig();
		config.setBrokerAddress("127.0.0.1:15555");
		Broker broker = new SingleBroker(config);
		final ServiceLoader serviceLoader = new ServiceLoader(broker);
		
		
		ServiceProvider sp = new HttpCallbackService(); 
		serviceLoader.loadService(sp);
	}

}
